/**
 * 
 */
package martinhynar.blog;

import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Logger;

import javax.xml.ws.handler.LogicalHandler;
import javax.xml.ws.handler.LogicalMessageContext;
import javax.xml.ws.handler.MessageContext;
import javax.xml.ws.handler.MessageContext.Scope;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * The Class ThroughputLimitHandler.
 */
public class ThroughputLimitHandler implements
		LogicalHandler<LogicalMessageContext> {

	/** Name of the system property that control maximum parallel requests. */
	private static final String REQUEST_LIMIT = "request.limit";

	/** Logging */
	private static final Logger LOG = Logger
			.getLogger(ThroughputLimitHandler.class.getName());

	/**
	 * The name for parameter that marks message with unique identifier so that
	 * it is obvious which message is processed.
	 */
	public static final String REQUEST_ID = "REQUEST_ID";

	/**
	 * The name for parameter that indicates state of traffic limitation to web
	 * service endpoint.
	 */
	public static final String LIMIT_REACHED = "LIMIT_REACHED";

	/** Default for maximum parallel requests. */
	private static final int DEFAULT_MAX_PARALLEL_REQUESTS = 5;

	/** The minimum tolerable value for maximum parallel requests. */
	private static final int MINIMUM_MAX_PARALLEL_REQUESTS = 2;

	/** The actual maximum parallel requests. */
	private int maxParallelRequests = DEFAULT_MAX_PARALLEL_REQUESTS;

	/** The actual parallel requests. */
	private AtomicInteger actualParallelRequests;

	/** The lock that guards changes to actual parallel requests. */
	private final Lock lock;

	/**
	 * Instantiates a new throughput limit handler.
	 */
	public ThroughputLimitHandler() {
		actualParallelRequests = new AtomicInteger(0);
		lock = new ReentrantLock();
		String limit = System.getProperty(REQUEST_LIMIT);
		try {
			int max = Integer.parseInt(limit);
			if (max < MINIMUM_MAX_PARALLEL_REQUESTS) {
				maxParallelRequests = DEFAULT_MAX_PARALLEL_REQUESTS;
				LOG.info(String
						.format("Found property %s set to %d. This value in under tolerable minimum. Using default value (%d) for maximum concurrent request limit.",
								REQUEST_LIMIT, limit,
								DEFAULT_MAX_PARALLEL_REQUESTS));
			} else {
				maxParallelRequests = max;
				LOG.info(String
						.format("Found property %s set to %d. Using this value as maximum concurrent request limit.",
								REQUEST_LIMIT, limit));
			}
		} catch (NumberFormatException nfex) {
			LOG.warning(String
					.format("Found property %s set to %s. This is not integer value, using default value %d",
							REQUEST_LIMIT, limit, DEFAULT_MAX_PARALLEL_REQUESTS));
		}
	}

	/**
	 * Handle message.
	 * 
	 * @param context
	 *            the context
	 * @return true, if successful
	 * 
	 * @see javax.xml.ws.handler.Handler#handleMessage(javax.xml.ws.handler.MessageContext)
	 */
	@Override
	public boolean handleMessage(LogicalMessageContext context) {
		// Get the direction of the message
		Boolean outbound = (Boolean) context
				.get(LogicalMessageContext.MESSAGE_OUTBOUND_PROPERTY);

		if (outbound) {
			// Response, decrease counter only
			try {
				lock.lock(); // block until condition holds
				int actual = actualParallelRequests.decrementAndGet();
				// Get identification of the message
				Object uuid = context.get(REQUEST_ID);
				LOG.info(String
						.format("[O: %s] Throughput state updated. [maximum = %d, actual = %d]",
								uuid, maxParallelRequests, actual));
			} finally {
				lock.unlock();
			}
		} else {
			// Request, do checks
			lock.lock(); // block until condition holds
			try {
				// Add internal identification of the message
				UUID uuid = UUID.randomUUID();
				context.put(REQUEST_ID, uuid);
				context.setScope(REQUEST_ID, Scope.APPLICATION);

				if (actualParallelRequests.get() < maxParallelRequests) {
					// We could pass the request further for processing
					LOG.info(String
							.format("[I: %s] Throughput not reached. [maximum = %d, actual = %d]",
									uuid, maxParallelRequests,
									actualParallelRequests.get()));
					context.put(LIMIT_REACHED, Boolean.FALSE);
					context.setScope(LIMIT_REACHED, Scope.APPLICATION);

				} else {
					context.put(LIMIT_REACHED, Boolean.TRUE);
					context.setScope(LIMIT_REACHED, Scope.APPLICATION);
					LOG.info(String
							.format("[I: %s] Throughput maximum reached. [maximum = %d, actual = %d]. Rejecting request.",
									uuid, maxParallelRequests,
									actualParallelRequests.get()));
				}

				// Increment counter, it must be known how many requests are
				// actually handled
				actualParallelRequests.incrementAndGet();

			} finally {
				lock.unlock();
			}
		}
		return true;
	}

	/**
	 * Handle fault.
	 * 
	 * @param context
	 *            the context
	 * @return true, if successful
	 * @see javax.xml.ws.handler.Handler#handleFault(javax.xml.ws.handler.MessageContext)
	 */
	@Override
	public boolean handleFault(LogicalMessageContext context) {
		// Failure response, decrease counter only
		try {
			lock.lock(); // block until condition holds
			int actual = actualParallelRequests.decrementAndGet();
			Object uuid = context.get(REQUEST_ID);
			LOG.info(String
					.format("[F: %s] Throughput state updated. [maximum = %d, actual = %d]",
							uuid, maxParallelRequests, actual));
		} finally {
			lock.unlock();
		}
		return true;
	}

	/**
	 * Close.
	 * 
	 * @param context
	 *            the context
	 * @see javax.xml.ws.handler.Handler#close(javax.xml.ws.handler.MessageContext)
	 */
	@Override
	public void close(MessageContext context) {
	}
}
